package com.bamboowind.producer.confirm;

import com.bamboowind.util.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;


/**
 * 发布确认
 *
 * @Author cpw
 * @Date 2022/11/1 11:01
 */
public class ConfirmWork {

    /**
     * 传递的消息数量
     */
    public static final Integer MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws Exception {
        // publishMessageIndividually();
        // publishMessageBatch();
        publishMessageAsync();
    }

    /**
     * 单个确认发布
     * 发布:1000个单独确认时间耗时:428ms
     */
    public static void publishMessageIndividually() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        // 随机队列名字
        String queueName = UUID.randomUUID().toString();
        // 创建队列
        channel.queueDeclare(queueName, false, false, false, null);
        // 开启发布确认
        channel.confirmSelect();
        // 开始时间
        long beginTime = System.currentTimeMillis();
        for (Integer i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes());
            // 同步确认发布
            boolean flag = channel.waitForConfirms();
            if (flag) {
                System.out.println("消息发送成功");
            }
        }
        // 结束时间
        long endTime = System.currentTimeMillis();
        System.out.println("发布:" + MESSAGE_COUNT + "个单独确认时间耗时:" + (endTime - beginTime) + "ms");
    }

    /**
     * 批量确认发布
     * 发布:1000个批量确认时间耗时:50ms
     */
    public static void publishMessageBatch() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);
        channel.confirmSelect();
        // 批量确认信息个数
        Integer batchSize = 100;
        long beginTime = System.currentTimeMillis();
        for (Integer i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes());
            // 每100次确认
            if (i % batchSize == 0) {
                channel.waitForConfirms();
            }
        }
        long endTime = System.currentTimeMillis();
        System.out.println("发布:" + MESSAGE_COUNT + "个批量确认时间耗时:" + (endTime - beginTime) + "ms");
    }

    /**
     * 异步处理
     * 发布:1000个异步确认消息耗时:39ms
     */
    public static void publishMessageAsync() throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);
        // 开启发布确认
        channel.confirmSelect();

        // 定义用于处理消息的安全队列，适用于高并发的情况
        ConcurrentSkipListMap<Long, Object> outstandingConfirms = new ConcurrentSkipListMap<>();

        /**
         * 消息确认成功的回调
         * 参数:deliveryTag:消息标识
         *      multiple:是否批量确认
         * */
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            // 删除掉成功的消息，剩下的就是没有成功的
            if (multiple) {
                // 从key等于null截取到key为指定的deliveryTag标识
                ConcurrentNavigableMap<Long, Object> confirmed = outstandingConfirms.headMap(deliveryTag);
                confirmed.clear();
            } else {
                outstandingConfirms.remove(deliveryTag);
            }
            System.out.println("确认的消息标记:" + deliveryTag);
        };

        //消息取消确认的回调
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            String usMessage = String.valueOf(outstandingConfirms.get(deliveryTag));
            System.out.println("没有确认的消息标记:" + deliveryTag + ",消息是:" + usMessage);
        };

        // 添加异步监听器
        channel.addConfirmListener(ackCallback, nackCallback);

        long beginDate = System.currentTimeMillis();
        for (Integer i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            // 将消息存入上面安全队列
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
            channel.basicPublish("", queueName, null, message.getBytes());
        }
        long endDate = System.currentTimeMillis();
        System.out.println("发布:" + MESSAGE_COUNT + "个异步确认消息耗时:" + (endDate - beginDate) + "ms");
    }
}
